package org.huangrui.spark.scala.streaming

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.receiver.Receiver

import scala.util.Random

/**
 * @Author hr
 * @Create 2024-10-21 17:16 
 */
object SparkStreaming02_DIY {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val sc: StreamingContext = new StreamingContext(sparkConf, Seconds(3))

    val messageDS: ReceiverInputDStream[String] = sc.receiverStream(new MyReceiver())
    messageDS.print()

    sc.start()
    sc.awaitTermination()
  }

  /**
   * 自定义数据采集器
   *     1. 继承Receiver，定义泛型, 传递参数
   *        2. 重写方法
   */
  class MyReceiver extends Receiver[String](StorageLevel.MEMORY_ONLY) {
    private var flag = true

    override def onStart(): Unit = {
      new Thread(new Runnable {
        override def run(): Unit = {
          while (flag) {
            val message: String = new Random().nextInt(10).toString
            store(message)
            Thread.sleep(500)
          }
        }
      }).start()
    }

    override def onStop(): Unit = {
      flag = false
    }
  }
}
